package uic.edu.dempsey;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.StringTokenizer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class Task3 {
    public static class MapperForWord extends Mapper<LongWritable, Text, Text, LongWritable> {
        private final static LongWritable one = new LongWritable(1);

        //private Text word=new Text();
        public String modify(String str) {
            String temp = str.replaceAll("//W", ""); //  //W means the non-character
            return temp;
        }

        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] words = line.split(" ");
            for (String word : words) {
                String temp;
                word=word.toLowerCase();  //transfer to lower case
                temp = modify(word);  //eliminate the notation
                temp=temp.replaceAll("[-_=|:;'`!?,.\"?~#%*&(){}]","");  //eliminate again
                //    PorterStemmer porterStemmer = new PorterStemmer();
                //    porterStemmer.setCurrent(temp);
                //   temp = porterStemmer.getCurrent();
                context.write(new Text(temp), one);
            }
        }

    }

    public static class SortByOccurence extends Mapper<LongWritable,Text,LongWritable,Text>
    {
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
        {
            int i=0;
            String temp="";
            int number=-100000;
            Pattern pattern= Pattern.compile("[0-9]+");
            StringTokenizer str=new StringTokenizer(value.toString());
            while(str.hasMoreElements())
            {
                String word=str.nextToken();
                i++;
                Matcher matcher=pattern.matcher(word);
                if(i==1)   //because every time it read a line which have number and word, so when the i equal to one, then it must be word, otherwise it would be number
                {
                    temp=word;
                }
                else if(i==2) {
                    number = Integer.parseInt(word);
                    context.write(new LongWritable(number), new Text(temp));  //finally write it into the memory
                    i = 0;
                    continue;
                }
            }

            }
        }

        public static class DecreasingSort extends LongWritable.Comparator{
        public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2)
        {
            return -super.compare(b1,s1,l1,b2,s2,l2);  //only override the return type.
        }
        }


    public static class SumForKey extends Reducer<Text,LongWritable,Text,LongWritable> {
        private LongWritable sumAll=new LongWritable();
        public void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException,InterruptedException{
            long sum=0;
            for(LongWritable value:values){
                sum=sum+value.get();  // easily do the sum for the key.
            }
            sumAll.set(sum);
            context.write(key,sumAll);
        }
    }

    public static class SortByNumber extends Reducer<LongWritable,Text,LongWritable,Text>
    {
        public void reduce(LongWritable key, Iterable<Text> values, Context context) throws IOException,InterruptedException
        {
            for(Text value:values)
            {
                context.write(key,value);
            }
        }
    }

    public static void main(String[] args) throws Exception{
        long startTime = System.currentTimeMillis();  //start time
        //________________________________________________
        //the first job is to calculate the frequency of the character.
        Configuration conf = new Configuration();
        Job job= Job.getInstance(conf,"word count");
        job.setJarByClass(Task3.class);
        FileInputFormat.addInputPath(job,new Path(args[0]));

        Path outputPath = new Path(args[1]);
        FileSystem fileSystem =FileSystem.get(conf);
        if (fileSystem.exists(outputPath)){
            fileSystem.delete(outputPath,true);
            System.out.println("outputPath: "+args[1]+" exists, but has been deleted.");
        }
        //Mapper information
        job.setMapperClass(Task3.MapperForWord.class);  //the first job is to calculate the frequency
        job.setMapOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        //Reducer information
        job.setReducerClass(Task3.SumForKey.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        FileOutputFormat.setOutputPath(job,new Path(args[1]));
        //__________________________________________________________
        //the second job is to do the increasing sort

        Job job2=Job.getInstance(conf,"Sort By occurence"); //use the frequency of the first job
        job2.setJarByClass(Task3.class);
        FileInputFormat.addInputPath(job2,new Path(args[1]));

        Path outputPath2 = new Path(args[2]);
        FileSystem fileSystem2 =FileSystem.get(conf);
        if (fileSystem2.exists(outputPath2))
        {
            fileSystem2.delete(outputPath2,true);
            System.out.println("outputPath: "+args[2]+" exists, but has been deleted.");
        }

        job2.setMapperClass(Task3.SortByOccurence.class);
        job2.setMapOutputKeyClass(LongWritable.class);
        job2.setOutputValueClass(Text.class);

        job2.setReducerClass(Task3.SortByNumber.class);
        job2.setOutputKeyClass(LongWritable.class);
        job2.setOutputValueClass(Text.class);

        FileOutputFormat.setOutputPath(job2,new Path(args[2]));

        //________________________________________________________________________________________________
        //the third job is to do the decreasing sort.

        Job job3=Job.getInstance(conf,"decreasing sort");  //use the frequency of the first job but with override function which can implement decreasing sort.
        job3.setJarByClass(Task3.class);
        FileInputFormat.addInputPath(job3,new Path(args[1]));

        Path outputPath3=new Path(args[3]);
        FileSystem fileSystem3=FileSystem.get(conf);
        if(fileSystem3.exists(outputPath3))
        {
            fileSystem3.delete(outputPath3,true);
            System.out.println("outputPath: "+args[2]+" exists, but has been deleted.");
        }
        job3.setMapperClass(Task3.SortByOccurence.class);
        job3.setMapOutputKeyClass(LongWritable.class);
        job3.setOutputValueClass(Text.class);
        job3.setSortComparatorClass(DecreasingSort.class);  //override the class so we can do the decreasing sort
        job3.setReducerClass(Task3.SortByNumber.class);
        job3.setOutputKeyClass(LongWritable.class);
        job3.setOutputValueClass(Text.class);
        FileOutputFormat.setOutputPath(job3,new Path(args[3])); //output the format to another dierctory.

        if(job.waitForCompletion(true)&&job2.waitForCompletion(true)) {
            long endTime = System.currentTimeMillis();
            System.out.println("---------------------Time use:"+(endTime-startTime)+"ms");
            System.exit(job3.waitForCompletion(true) ? 0 : 1);
        }
        }
}
